package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class Demo3DIANXIN {
    Connection conn;
    Admin admin; // 对表结构的修改或者增加删除表
    Table dianxin; // 对表中的数据进行操作
    TableName dianxinT = TableName.valueOf("DIANXIN");

    @Before
    public void init() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,node1,node2");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        dianxin = conn.getTable(dianxinT);

    }

    @Test
    // 建表DIANXIN 列簇：cf1
    public void createTable() throws IOException {
        if (!admin.tableExists(dianxinT)) {
            admin.createTable(new HTableDescriptor(dianxinT).addFamily(new HColumnDescriptor("cf1").setMaxVersions(3)));
        }

    }

    @Test
    // 将数据写入HBase，以mdn作为rowkey，以起始时间作为写入hbase时的时间戳
    public void putAll() throws IOException {
        BufferedReader br = new BufferedReader(new FileReader("data/DIANXIN.csv"));
        String line = null;
        int batchSize = 1000;
        ArrayList<Put> puts = new ArrayList<>();
        while ((line = br.readLine()) != null) {
            // 批量写入

            String[] split = line.split(",");
            byte[] mdn = split[0].getBytes();
            long start_time = Long.parseLong(split[1]); // 以start_time作为每条数据的时间戳
            byte[] lg = split[4].getBytes(); // 经度
            byte[] lat = split[5].getBytes(); // 维度
            Put put = new Put(mdn);// 以mdn作为rowkey
            put.addColumn("cf1".getBytes(), "lg".getBytes(), start_time, lg);
            put.addColumn("cf1".getBytes(), "lat".getBytes(), start_time, lat);
            puts.add(put);
            if (puts.size() == batchSize) {
                dianxin.put(puts);
                puts = new ArrayList<Put>();
            }


        }
        if (puts.size() != 0) {
            dianxin.put(puts);
        }
    }

    @Test
    // 根据用户的mdn获取用户最新的3个位置
    public void getPosition() throws IOException {
        String mdn = "FDA42AB4AA84D13C2AC7512D1BAEE93B7DCC327A";

        Get get = new Get(mdn.getBytes());
        get.setMaxVersions(3);
        Result rs = dianxin.get(get);
        // 只能取到最新的一条记录
//        String lg = Bytes.toString(rs.getValue("cf1".getBytes(), "lg".getBytes()));
//        String lat = Bytes.toString(rs.getValue("cf1".getBytes(), "lat".getBytes()));
//        System.out.println(lg + "," + lat);
        ArrayList<String> lgList = new ArrayList<>();
        ArrayList<String> latList = new ArrayList<>();
        // 要取多版本的数据 需要使用CellUtil这种方式获取
        for (Cell cell : rs.listCells()) {
            String qua = Bytes.toString(CellUtil.cloneQualifier(cell));// 列名
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            if ("lg".equals(qua)) {
                lgList.add(value);
            } else if ("lat".equals(qua)) {
                latList.add(value);
            }
        }

        // 格式化输出
        for (int i = 0; i < 3; i++) {
            System.out.println(lgList.get(i) + "," + latList.get(i));
        }


    }


    @After
    public void closed() throws IOException {
        conn.close();
    }
}
